package com.zx.notification.producer.mq;


import com.zx.notification.common.constants.NotificationConstants;
import com.zx.notification.common.dao.NotificationChannelsConfigMapper;
import com.zx.notification.common.enums.MqTypeEnum;
import com.zx.notification.common.model.NotificationChannelsConfig;
import com.zx.notification.common.model.NotificationPushTasks;
import com.zx.notification.producer.config.ActiveMqProducerConfig;
import com.zx.notification.producer.config.KafkaProducerConfig;
import com.zx.notification.producer.config.RabbitMqProducerConfig;
import com.zx.notification.producer.config.RocketMqProducerConfig;
import com.zx.redis.block.service.RedisBlockService;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.util.concurrent.TimeUnit;

/**
 * MQ推送服务
 * 根据不同的MQ类型执行相应的推送逻辑
 */
@Slf4j
@Service
public class MqPushService {

    // Redis缓存相关常量
    private static final String CHANNEL_CONFIG_CACHE_KEY_PREFIX = NotificationConstants.RedisCache.CHANNEL_CONFIG_CACHE_KEY_PREFIX;
    private static final long CHANNEL_CONFIG_CACHE_EXPIRE_TIME = NotificationConstants.RedisCache.CHANNEL_CONFIG_CACHE_EXPIRE_TIME;
    private final NotificationChannelsConfigMapper notificationChannelsConfigMapper;
    private final ActiveMqProducerConfig activeMqConfig;
    private final RabbitMqProducerConfig rabbitMqConfig;
    private final KafkaProducerConfig kafkaMqConfig;
    private final RocketMqProducerConfig rocketMqConfig;


    private final RabbitTemplate rabbitTemplate;
    private final KafkaTemplate<String, String> kafkaTemplate;
    private final RocketMQTemplate rocketMQTemplate;
    private final JmsTemplate jmsTemplate;
    private final RedisBlockService redisBlockService;
    private String rabbitmqExchange;
    private String rabbitmqRoutingKey;
    private String kafkaTopic;
    private String rocketmqTopic;
    private String rocketmqTag;

    public MqPushService(NotificationChannelsConfigMapper notificationChannelsConfigMapper,
                         ActiveMqProducerConfig activeMqConfig,
                         RabbitMqProducerConfig rabbitMqConfig,
                         KafkaProducerConfig kafkaMqConfig,
                         RocketMqProducerConfig rocketMqConfig,
                         @Autowired(required = false) RabbitTemplate rabbitTemplate,
                         @Autowired(required = false) KafkaTemplate<String, String> kafkaTemplate,
                         @Autowired(required = false) RocketMQTemplate rocketMQTemplate,
                         @Autowired(required = false) JmsTemplate jmsTemplate,
                         RedisBlockService redisBlockService) {
        this.notificationChannelsConfigMapper = notificationChannelsConfigMapper;
        this.activeMqConfig = activeMqConfig;
        this.rabbitMqConfig = rabbitMqConfig;
        this.kafkaMqConfig = kafkaMqConfig;
        this.rocketMqConfig = rocketMqConfig;
        this.rabbitTemplate = rabbitTemplate;
        this.kafkaTemplate = kafkaTemplate;
        this.rocketMQTemplate = rocketMQTemplate;
        this.jmsTemplate = jmsTemplate;
        this.redisBlockService = redisBlockService;
    }

    @PostConstruct
    public void init() {
        log.info("MQ开关配置: RabbitMQ={}, Kafka={}, RocketMQ={}, ActiveMQ={}",
                rabbitMqConfig.isRabbitmqEnabled(), kafkaMqConfig.isKafkaEnabled(), rocketMqConfig.isRocketmqEnabled(), activeMqConfig.isEnabled());
        log.info("MQ模板Bean是否存在: RabbitTemplate={}, KafkaTemplate={}, RocketMQTemplate={}, JmsTemplate={}",
                rabbitTemplate != null, kafkaTemplate != null, rocketMQTemplate != null, jmsTemplate != null);

        // 添加提示信息，指导用户如果模板Bean不存在应该检查哪些地方
        if (rabbitTemplate == null || kafkaTemplate == null || rocketMQTemplate == null || jmsTemplate == null) {
            log.info("提示：如果某些MQ模板Bean不存在，请检查以下几点：");
            log.info("1. 对应的MQ服务是否已安装并运行（如RabbitMQ、Kafka、RocketMQ、ActiveMQ等）");
            log.info("2. application.yml或Nacos配置中对应的MQ配置是否正确");
            log.info("3. 相关的starter依赖是否已添加到pom.xml中");
            log.info("4. 对于RocketMQ，检查是否配置了rocketmq.name-server属性");
            log.info("5. 对于ActiveMQ，检查是否配置了spring.activemq.broker-url属性");
        }
    }

    /**
     * 推送消息到RabbitMQ
     *
     * @param pushTask 推送任务
     * @param message  消息内容
     */
    public void pushToRabbitMQ(NotificationPushTasks pushTask, String message) {
        log.info("推送通知[{}]到RabbitMQ: {}", pushTask.getNotificationId(), message);
        try {
            if (!rabbitMqConfig.isRabbitmqEnabled()) {
                log.warn("RabbitMQ未启用，无法推送消息到RabbitMQ");
                throw new RuntimeException("RabbitMQ未启用，无法推送消息到RabbitMQ");
            }

            if (rabbitTemplate != null) {
                String routingKey = pushTask.getRoutingKey() != null ? pushTask.getRoutingKey() : rabbitmqRoutingKey;
                rabbitTemplate.convertAndSend(rabbitmqExchange, routingKey, message);
                log.info("通知[{}]已成功推送到RabbitMQ", pushTask.getNotificationId());
            } else {
                log.warn("RabbitTemplate未配置，无法推送消息到RabbitMQ");
                throw new RuntimeException("RabbitTemplate未配置，无法推送消息到RabbitMQ");
            }
        } catch (Exception e) {
            log.error("推送通知[{}]到RabbitMQ失败", pushTask.getNotificationId(), e);
            throw new RuntimeException("推送通知到RabbitMQ失败", e);
        }
    }

    /**
     * 推送消息到RocketMQ
     *
     * @param pushTask 推送任务
     * @param message  消息内容
     */
    public void pushToRocketMQ(NotificationPushTasks pushTask, String message) {
        log.info("推送通知[{}]到RocketMQ: {}", pushTask.getNotificationId(), message);
        try {
            if (!rocketMqConfig.isRocketmqEnabled()) {
                log.warn("RocketMQ未启用，无法推送消息到RocketMQ");
                throw new RuntimeException("RocketMQ未启用，无法推送消息到RocketMQ");
            }

            if (rocketMQTemplate != null) {
                String destination = rocketmqTopic + ":" + (pushTask.getTag() != null ? pushTask.getTag() : rocketmqTag);
                rocketMQTemplate.convertAndSend(destination, message);
                log.info("通知[{}]已成功推送到RocketMQ", pushTask.getNotificationId());
            } else {
                log.warn("RocketMQTemplate未配置，无法推送消息到RocketMQ");
                throw new RuntimeException("RocketMQTemplate未配置，无法推送消息到RocketMQ");
            }
        } catch (Exception e) {
            log.error("推送通知[{}]到RocketMQ失败", pushTask.getNotificationId(), e);
            throw new RuntimeException("推送通知到RocketMQ失败", e);
        }
    }

    /**
     * 推送消息到ActiveMQ
     *
     * @param pushTask 推送任务
     * @param message  消息内容
     */
    public void pushToActiveMQ(NotificationPushTasks pushTask, String message) {
        log.info("推送通知[{}]到ActiveMQ: {}", pushTask.getNotificationId(), message);
        try {
            if (!activeMqConfig.isEnabled()) {
                log.warn("ActiveMQ未启用，无法推送消息到ActiveMQ");
                throw new RuntimeException("ActiveMQ未启用，无法推送消息到ActiveMQ");
            }

            if (jmsTemplate != null) {
                jmsTemplate.convertAndSend(pushTask.getDestinationName(), message);
                log.info("通知[{}]已成功推送到ActiveMQ", pushTask.getNotificationId());
            } else {
                log.warn("JmsTemplate未配置，无法推送消息到ActiveMQ");
                throw new RuntimeException("JmsTemplate未配置，无法推送消息到ActiveMQ");
            }
        } catch (Exception e) {
            log.error("推送通知[{}]到ActiveMQ失败", pushTask.getNotificationId(), e);
            throw new RuntimeException("推送通知到ActiveMQ失败", e);
        }
    }

    /**
     * 推送消息到Kafka
     *
     * @param pushTask 推送任务
     * @param message  消息内容
     */
    public void pushToKafka(NotificationPushTasks pushTask, String message) {
        log.info("推送通知[{}]到Kafka: {}", pushTask.getNotificationId(), message);
        try {
            if (!kafkaMqConfig.isKafkaEnabled()) {
                log.warn("Kafka未启用，无法推送消息到Kafka");
                throw new RuntimeException("Kafka未启用，无法推送消息到Kafka");
            }

            if (kafkaTemplate != null) {
                String topic = pushTask.getDestinationName() != null ? pushTask.getDestinationName() : kafkaTopic;
                kafkaTemplate.send(topic, message);
                log.info("通知[{}]已成功推送到Kafka", pushTask.getNotificationId());
            } else {
                log.warn("KafkaTemplate未配置，无法推送消息到Kafka");
                throw new RuntimeException("KafkaTemplate未配置，无法推送消息到Kafka");
            }
        } catch (Exception e) {
            log.error("推送通知[{}]到Kafka失败", pushTask.getNotificationId(), e);
            throw new RuntimeException("推送通知到Kafka失败", e);
        }
    }

    /**
     * 根据推送任务信息推送消息到对应的MQ
     *
     * @param pushTask 推送任务
     * @param message  消息内容
     */
    public void pushMessage(NotificationPushTasks pushTask, String message) {
        // 获取渠道配置
        NotificationChannelsConfig channelConfig = getChannelConfigFromCache(pushTask.getChannelType());
        if (channelConfig == null) {
            log.warn("未找到渠道类型[{}]的配置信息", pushTask.getChannelType());
            throw new RuntimeException("未找到渠道类型[" + pushTask.getChannelType() + "]的配置信息");
        }

        // 根据MQ类型执行相应的推送处理
        switch (channelConfig.getMqType().toUpperCase()) {
            case MqTypeEnum.RABBITMQ_CODE:
                pushToRabbitMQ(pushTask, message);
                break;
            case MqTypeEnum.ROCKETMQ_CODE:
                pushToRocketMQ(pushTask, message);
                break;
            case MqTypeEnum.ACTIVEMQ_CODE:
                pushToActiveMQ(pushTask, message);
                break;
            case MqTypeEnum.KAFKA_CODE:
                pushToKafka(pushTask, message);
                break;
            default:
                log.warn("不支持的MQ类型: {}", channelConfig.getMqType());
                throw new RuntimeException("不支持的MQ类型: " + channelConfig.getMqType());
        }
    }

    /**
     * 从Redis缓存获取渠道配置，如果缓存中不存在则从数据库获取并存入缓存
     *
     * @param channelType 渠道类型
     * @return 渠道配置
     */
    private NotificationChannelsConfig getChannelConfigFromCache(String channelType) {
        String cacheKey = CHANNEL_CONFIG_CACHE_KEY_PREFIX + channelType;
        NotificationChannelsConfig config = redisBlockService.getCachedObject(cacheKey, NotificationChannelsConfig.class);
        if (config == null) {
            // 缓存中不存在，从数据库获取
            config = notificationChannelsConfigMapper.selectByChannelType(channelType);
            if (config != null) {
                // 存入缓存，设置过期时间
                redisBlockService.setCacheObject(cacheKey, config, CHANNEL_CONFIG_CACHE_EXPIRE_TIME, TimeUnit.SECONDS);
            }
        }
        return config;
    }
}